Skip to content

[SPARK-58089][SQL] Push variant extractions through Aggregate/Sort/Join#57190

Open
qlong wants to merge 1 commit into
apache:masterfrom
qlong:variant-pushout-aggregate
Open

[SPARK-58089][SQL] Push variant extractions through Aggregate/Sort/Join#57190
qlong wants to merge 1 commit into
apache:masterfrom
qlong:variant-pushout-aggregate

Conversation

@qlong

@qlong qlong commented Jul 10, 2026

Copy link
Copy Markdown
Contributor

What changes were proposed in this pull request?

Introduce a new optimizer rule PullOutVariantExtractions that hoists
variant_get / Cast(variant) extractions out of three operator types
that the existing PushVariantIntoScan / V2ScanRelationPushDown rules
cannot see through:

  • Aggregate function arguments – e.g. max(variant_get(v, '$.price', 'int')):
    the extraction is moved into a Project directly below the Aggregate and the
    aggregate references the resulting alias. The bare variant column is suppressed
    unless it is also needed raw (e.g. as a GROUP BY key), so no redundant
    full-variant slot is generated.

  • Sort order keys – e.g. ORDER BY variant_get(v, '$.price', 'int'):
    matched as Project → Sort; the extraction is hoisted below the Sort and
    the original Project is reproduced to prevent the alias from leaking into
    the output.

  • Join conditions and projections above joins – matched as Project → Join;
    extractions in both the join condition and the outer Project are routed to
    the owning join side. A pushSideAliases helper then pushes the aliases
    through any depth of chained joins so they land in a Project directly
    above the scan (where PhysicalOperation collapses them with the scan, making
    them visible to the pushdown). This is necessary because PhysicalOperation
    stops at a Join node.

A Sort sitting over a Join is handled by fusing the two cases: the
order-key aliases are pushed through the join tree, not left in a Project
above it.

The rule is gated by a new internal config
spark.sql.variant.pushVariantIntoScan.pullOutExtractions (default true)
and is a no-op unless spark.sql.variant.pushVariantIntoScan is also enabled.
Non-variant plans are untouched.

The rule is registered as the first rule in SparkOptimizer.earlyScanPushDownRules,
before SchemaPruning and the V2 scan pushdown rules.

Why are the changes needed?

Before this change, a variant_get inside an aggregate function argument, sort key,
or join condition caused the whole variant column to be read raw (or shredded with a
redundant full-variant slot). For example:

SELECT name, max(variant_get(v, '$.price', 'int')) FROM T GROUP BY name

read the entire v column even though only the price field was needed.
After this change, Spark shreds only the requested typed fields, avoiding the
full-variant I/O.

The change improves query performance for variant referenced in aggregrate, join, sort.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Added new units. Also run correctness tests against some known workload.

Performance result

The test run against tpc-ds dataset (SF=5). Run B is with the new pullout rule enabled.

Compare: tpcds-flat-20260710 vs tpcds-flat-pullout-20260710  (metric: query_median)
  Run A: /Users/qlong/opensources/variant-conformance-benchmark/results/tpcds-flat-20260710/tpcds-flat/timings-variant.csv
  Run B: /Users/qlong/opensources/variant-conformance-benchmark/results/tpcds-flat-pullout-20260710/tpcds-flat/timings-variant.csv

query    median_A   median_B   delta_s   delta_%
------   --------   --------   -------   -------
q07            7.41       4.45    +2.95   +66.3%
q12            1.64       1.66    -0.02    -1.3%
q19            2.21       2.15    +0.07    +3.1%
q26            8.90       4.39    +4.51  +102.8%
q42            5.83       3.51    +2.32   +66.0%
q52            7.42       4.55    +2.87   +63.0%
q55            5.84       3.60    +2.24   +62.3%
q63            6.91       3.78    +3.13   +82.8%
q68            7.02       4.87    +2.15   +44.2%
q73            6.55       4.27    +2.28   +53.4%
q79            6.53       5.00    +1.53   +30.7%
q98            6.62       4.55    +2.07   +45.4%

Summary: 12 queries, 12 comparable
  Geo-mean delta: +48.6%  (Run B faster)
  Total (query_median):   72.9s vs 46.8s

Was this patch authored or co-authored using generative AI tooling?

Co-authored with Claude Code

`PushVariantIntoScan` (v1) and `V2ScanRelationPushDown` (v2)
both rely on `PhysicalOperation`, which collapses only a
contiguous `Project`/`Filter` chain and stops at `Aggregate`,
`Sort`, and `Join`. As a result, `variant_get` expressions
embedded in aggregate function arguments, sort-order
expressions, or join conditions are invisible to the pushdown
rules, and the full variant column is read raw instead of
being shredded to the requested typed fields.

When an `Aggregate` or `Sort` sits above a `Join`, the
barrier compounds: even a `Project` hoisted above the join
tree is still unreachable from the scan side.

Example queries that fail to shred without this fix:

  -- Aggregate: extraction in agg arg / GROUP BY
  SELECT AVG(variant_get(data, '$.qty', 'double'))
  FROM   store_sales
  GROUP  BY variant_get(data, '$.id', 'string')

  -- Join: extraction in ON condition
  SELECT ss.k
  FROM   store_sales ss
  JOIN   date_dim d
    ON ss.date_sk = variant_get(d.data, '$.sk', 'int')

  -- Aggregate over Join (TPC-DS Q26 shape)
  SELECT AVG(variant_get(ss.data, '$.qty', 'double'))
  FROM   store_sales ss
  JOIN   date_dim  d  ON ss.date_sk  = d.date_sk
  JOIN   store     st ON ss.store_sk = st.store_sk
  GROUP  BY variant_get(ss.data, '$.id', 'string')

In all cases the scan emits `data:struct<0:variant>` (full
blob) instead of `data:struct<0:double, 1:string>`.

Introduce a new optimizer rule `PullOutVariantExtractions`,
registered as the first rule in `earlyScanPushDownRules`
(before `SchemaPruning`), which hoists `variant_get` calls
out of `Aggregate`, `Sort`, and `Join` into a `Project`
directly below the operator so the downstream pushdown rules
can see them.

- **Aggregate**: hoist extractions from aggregate function
  arguments into a `Project` below the `Aggregate`. The
  `Aggregate` defines its own output so the raw variant
  column is not passed through.

- **Sort / Join**: match the `Project` sitting directly above
  the `Sort`/`Join` (its `references` give the live-above
  set) and drop any variant column no longer referenced once
  its extraction is hoisted.

- **Push-through-Join**: hoisting above a `Join` is not
  enough because `PhysicalOperation` stops there. The new
  `pushSideAliases` helper recursively routes each hoisted
  `_ve` alias down through the join tree to the side whose
  output owns the referenced attribute, landing it in a
  `Project` directly above the scan. Handles any depth of
  chained joins in one pass.for review
@qlong qlong force-pushed the variant-pushout-aggregate branch from 16b2f4b to ef2a92e Compare July 11, 2026 04:49
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant